/*
 * Copyright (c) 2017, 1DAOYUN and/or its affiliates. All rights reserved.
 * 1DAOYUN PROPRIETARY/CONFIDENTIAL. Use is subject to license terms.
 */
package com.iot.nev.gateway.client.kafka;

import java.io.IOException;
import java.util.Collections;
import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.log4j.Logger;

import com.iot.nev.gateway.client.Client;
import com.iot.nev.gateway.connector.Connector;
import com.iot.nev.gateway.connector.hbase.SaveDataForHBase;
import com.iot.nev.gateway.connector.mongdb.SaveDataForMongoDB;
import com.iot.nev.gateway.connector.mysql.SaveDataForMysql;
import com.iot.nev.gateway.utils.UtilTools;

import kafka.utils.ShutdownableThread;

/**
 * Kafka路由输出，调用Connector接口，进行保存或处理。
 * 
 * @author Administrator
 *
 */
public class KafkaConsumerClient extends ShutdownableThread implements Client{

	/** Consumer object */
	private KafkaConsumer<Integer, String> consumer;
	/** logger */
	private Logger logger = Logger.getLogger(KafkaConsumerClient.class);
	/** config */
	private Properties config = UtilTools.getConfig(System.getProperty("user.dir") + "/configuration/config.properties");

	private String topic = "Tina-topic";
	
	public KafkaConsumerClient() {
		super("KafkaConsumerRouter", false);
		init();
	}

	/**
	 * 初始化.
	 */
	public void init() {
		this.topic = config.getProperty("router_kafka_server_topic");
		boolean isAutoCommit = Boolean.valueOf(config.getProperty("router_kafka_server_comsumer_auto_commit"));
		String autoCommitInterval = config.getProperty("router_kafka_server_comsumer_auto_commit_interval");
		String sessionTimeout = config.getProperty("router_kafka_server_comsumer_session_timeout");
		String serverUrl = config.getProperty("router_kafka_server_url");
		String serverPort = config.getProperty("router_kafka_server_port");
		
		Properties props = new Properties();
		props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, serverUrl + ":" + serverPort);
		props.put(ConsumerConfig.GROUP_ID_CONFIG, getID());
		props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, isAutoCommit);
		props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, autoCommitInterval);
		props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, sessionTimeout);
		props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.IntegerDeserializer");
		props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
		consumer = new KafkaConsumer<>(props);
		
	}
	
	int num = 1;
	
	@Override
	public void doWork() {
		consumer.subscribe(Collections.singletonList(this.topic));
		ConsumerRecords<Integer, String> records = consumer.poll(1000);
		// 消费
		for (ConsumerRecord<Integer, String> record : records) {
			try {
				System.out.println(num++ + "--" + System.currentTimeMillis());
				logger.info(num);
				// SaveDataForHBase.saveData(record.value());
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
	}

	@Override
	public boolean isInterruptible() {
		return false;
	}

	/**
	 * 每个一个ID.
	 * 
	 * @return
	 */
	public String getID() {
		return KafkaConsumerClient.class.getName() + System.currentTimeMillis();
	}

	@Override
	public Connector getConnector() {
		// TODO Auto-generated method stub
		return null;
	}

	@Override
	public Client getClient() {
		// TODO Auto-generated method stub
		return null;
	}

	@Override
	public void restart() {
		// TODO Auto-generated method stub
		
	}

	@Override
	public boolean isActive() {
		// TODO Auto-generated method stub
		return false;
	}

	@Override
	public int getConnectedCount() {
		// TODO Auto-generated method stub
		return 0;
	}

	@Override
	public float getServiceLevel() {
		// TODO Auto-generated method stub
		return 0;
	}
	
}
